package com.ly.kafka.config;

import com.ly.kafka.empty.Order;
import com.ly.kafka.empty.OrderDTO;
import com.ly.kafka.pool.KafkaConsumerPool;
import com.ly.kafka.service.OrderService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import javax.annotation.PostConstruct;

@Configuration
public class KafkaConsumerPoolConfig {


    //一个业务需要一个线程池
    @Autowired
    private OrderService orderService;


    /**
     * order业务顺序消费池
     */
    private KafkaConsumerPool<Order> kafkaConsumerPool;

    /**
     * 顺序消费并发级别
     */
    @Value("${kafka-consumer-pool-params.concurrent}")
    private Integer concurrent;

    /**
     * @describe：初始化顺序消费池
     * @author：李勇
     * @date：2023/5/24
     * @param
     * @return：void
     */
    @Bean
    public KafkaConsumerPool orderConsumerPool(){
        KafkaSortConsumerConfig<OrderDTO> config = new KafkaSortConsumerConfig<>();
        config.setBizName("order");
        // java的一个功能接口 用来作为lambda表达式或方法引用的任务目标(传递一个参数执行指定的方法)
        //将处理任务的业务层注入进去，处理任务
        config.setBizService(order -> {
            try {
                orderService.solveRetry(order);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        });
        config.setConcurrentSize(concurrent);
        return new KafkaConsumerPool<>(config);
    }




}
